<?php

namespace app\command;

use app\service\GatewayWorkerService;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\console\input\Option;
use app\common\model\UserModel;
use app\common\model\RegSourceModel;
use app\common\model\StatPrivateChatModel;

class RabbitPush extends Command
{
    protected function configure()
    {
        $this->setName('admrabbit')
            ->addOption('mode', 'm', Option::VALUE_OPTIONAL, '模式')
            ->setDescription('MQ 推送');
    }

    protected function execute(Input $input, Output $output)
    {
        if ($input->hasOption('mode')) {
            $mode = trim($input->getOption('mode'));
        } else {
            $mode = 'private_chat';
        }

        if (!in_array($mode, ['private_chat', 'group_chat'])) {
            $output->writeln("mode参数 错误");
        }

        switch ($mode) {
            case 'private_chat':
                $this->commonExecute($mode, 'privateChatPushMessage');
                break;
            case 'group_chat':
                $this->commonExecute($mode, 'groupChatPushMessage');
                break;
        }

    }

    protected function commonExecute($mode, $func)
    {
        $param = config('rabbitmq.');
        if ($connection = $this->connection($param)) {
            try {
                $channel = $connection->channel();
            } catch (\Exception $exception) {
                throw $exception;
            }

            try {
                $channel->basic_qos(0, 1, false);
                $channel->queue_declare($param[$mode]['queue_name'], false, true, false, false);
                $channel->exchange_declare($param[$mode]['exchange_name'], $param[$mode]['exchange_type'], false, true, false);
                $channel->queue_bind($param[$mode]['queue_name'], $param[$mode]['exchange_name'], $param[$mode]['route_key']);
                $channel->basic_consume($param[$mode]['queue_name'], $param[$mode]['consumer_tag'], false, true, false, false, [$this, $func]);

                while (true) {
                    if (count($channel->callbacks)) {
                        $channel->wait();
                    }
                }
            } catch (\Exception $exception) {
                $channel->close();
                $connection->close();

                return $this->commonExecute($mode, $func);
            }
        }
    }

    protected function connection(array $param, int $num = 0)
    {
        $num++;
        try {
            if ($num > 5) {
                return null;
            }

            return new AMQPStreamConnection(
                $param['host'],
                $param['port'],
                $param['login'],
                $param['password'],
                $param['vhost']
            );

        } catch (\Exception $exception) {
            // 写错误日志
            writeLog($exception->getMessage(), 'rabbit_push_data.txt');
            return $this->connection($param, $num);
        }
    }

    /**
     * @param AMQPMessage $message
     */
    public function privateChatPushMessage(AMQPMessage $message)
    {
        $send_msg = json_decode($message->body, true);
        if (empty($send_msg)) {
            return;
        }

        $send_msg['chat_type'] = 1;
        // 写发送日志
         writeLog($send_msg, 'rabbit_push_data.txt');
        try {
            $send_msg['user_id'] = (string)$send_msg['sender_uid'];
            $send_msg['anchor_id'] = $send_msg['receive_uid'];
            //add stat private chat log
            $this->addStatPrivateChat($send_msg['sender_uid'],$send_msg['receive_uid']);
            // 推送消息到socket
            GatewayWorkerService::sendMsg('say', $send_msg);
        } catch (\Exception $exception) {
            // 写错误日志
            writeLog($exception->getMessage().$exception->getFile().$exception->getLine(), 'rabbit_push_data.txt');
        }
    }

    /**
     * @param AMQPMessage $message
     */
    public function groupChatPushMessage(AMQPMessage $message)
    {
        $boby = $result = str_replace('"seq":', '"seq":"', $message->body);
        $boby = str_replace(',"uid"', '","uid"', $boby);

        $send_msg = json_decode($boby, true);
        if (empty($send_msg)) {
            return;
        }
        if ($send_msg['type'] == 3 &&  is_string($send_msg['content'])) {
            $send_msg['content'] = json_decode($send_msg['content'], true) ?? [];
        }
        $send_msg['chat_type'] = 2;

        $explode = explode('_', $send_msg['group_id']);
        if (count($explode) == 2) {
            $send_msg['group_id'] = $explode[1];
        }

        try {
            // 推送消息到socket
            GatewayWorkerService::sendMsg('group_say', $send_msg);
        } catch (\Exception $exception) {
            // 写错误日志
            writeLog($exception->getMessage(), 'rabbit_push_data.txt');
        }
    }

    public function addStatPrivateChat(int $sender_uid = 0, int $receive_uid = 0) {
        if (!$send_user = UserModel::getInfo($sender_uid)) {
            return ;
        }

        if (!$receive_user = UserModel::getInfo($receive_uid)) {
            return ;
        }

        $regsource_info = RegSourceModel::where('source_key', $send_user['reg_source_id'])->find();
        if(empty($regsource_info)) {
            return ;
        }

        $model = new StatPrivateChatModel();
        $model->sender_uid = $sender_uid;
        $model->receive_uid = $receive_uid;
        $model->source_key = $regsource_info["source_key"];
        $model->source_name = $regsource_info["source_name"];
        $model->date = date("Y-m-d");
        $model->created_at = date("Y-m-d H:i:s");
        $model->save();

        return ;
    }
}
